package com.hirain.zookeeper.curator;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorEventType;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.RetryUntilElapsed;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.junit.Test;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by caicf on 2016/9/1.
 */
public class CurTest {

    @Test
    public void test() throws Exception {
        //重试策略    固定时间固定次数重试
//        RetryPolicy retryPolicy=new ExponentialBackoffRetry(1000,3);
        //最大重试次数，两次重试之间的时间间隔（毫秒）
//        RetryPolicy retryPolicy=new RetryNTimes(3,1000);
        //重试直到超过指定的时间，第二个参数两次重试之间的时间间隔（毫秒）
        RetryPolicy retryPolicy=new RetryUntilElapsed(10000,2000);
//        CuratorFramework client= CuratorFrameworkFactory.newClient("127.0.0.1:2181",1000,1000,retryPolicy);
        CuratorFramework client=CuratorFrameworkFactory
                .builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(1000)
                .connectionTimeoutMs(1000)
                .retryPolicy(retryPolicy)
                .build();
        client.start();

        //数据节点的删除
        String path=client.create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.EPHEMERAL)
                .forPath("/curtest/child1","hello curator".getBytes());
        /**
         * 删除节点
         * deletingChildrenIfNeeded()  如果有子节点删除
         * guaranteed()   持续在后台删除，保障节点的删除
         */
        client.delete()
                .guaranteed().deletingChildrenIfNeeded().withVersion(1).forPath("/curtest/child1");

        //获取子节点
        List<String> childList=client.getChildren().forPath("/curtest");

        //获取节点内容
        byte[] ret = client.getData().forPath("/curtest");
        //获取节点状态信息
        Stat stat=new Stat();
        byte[] data = client.getData().storingStatIn(stat).forPath("/curtest");

        //修改节点内容   如果需要校验版本信息withVersion()
        client.setData().withVersion(stat.getVersion()).forPath("/curtest","1111".getBytes());

        ExecutorService es= Executors.newFixedThreadPool(4);

        //判断节点是否存在  存在返回Stat对象，不存在返回空
        client.checkExists().inBackground(new BackgroundCallback() {
            /**
             * 异步调用
             * 使用.inBackground()
             */
            @Override
            public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
                CuratorEventType eventType=curatorEvent.getType();
                int resultCode=curatorEvent.getResultCode();//成功返回0，失败返回非0
                //上下文信息，异步调用时传入  如“123”
                Object o=curatorEvent.getContext();
                curatorEvent.getPath();  //得到节点路径
                curatorEvent.getChildren();  //子节点列表
                curatorEvent.getData();   //节点数据
            }
        },"123",es).forPath("/curtest");

        /**
         * 监听节点
         */
        final NodeCache cache = new NodeCache(client, "/curtest");
        cache.start(); //开启监听
        //注册监听
        cache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                byte [] ret=cache.getCurrentData().getData(); //得到当前节点的最新数据
            }
        });


        /**
         * 监听子节点
         */
        final PathChildrenCache cache1 = new PathChildrenCache(client, "/curtest",true);
        cache1.start();
        cache1.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                switch (pathChildrenCacheEvent.getType()){
                    case CHILD_ADDED:
                        //获取节点信息
                        System.out.println(pathChildrenCacheEvent.getData().getPath());
                        break;
                    case CHILD_UPDATED:
                        break;
                    case CHILD_REMOVED:
                        break;
                    default:
                        break;
                }
            }
        });
    }
}
